package com.blackman.skcrawler.core.zookeeper;

import com.alibaba.fastjson.JSON;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;

import java.lang.management.ManagementFactory;
import java.lang.management.ThreadInfo;
import java.lang.management.ThreadMXBean;
import java.util.List;

/**
 * 分布式锁
 *
 * @author blackman
 * @date 2018/04/21 22:41
 *
 */
public class DistributedLock {

    private static String servers = "192.168.3.177:2181";

    private static String LOCK_PARENT_PATH = "/sys/app_lock";

    private static ZkClient zkClient;

    static{
        zkClient = new ZkClient(servers, 10000);
        boolean parentExist = existLock(LOCK_PARENT_PATH);
        if(!parentExist){
            zkClient.createPersistent(LOCK_PARENT_PATH, true);
        }
    }

    /**
     * 判断文件目录是否存在
     * @param path
     * @return
     */
    public static boolean existLock(String path){
        return zkClient.exists(path);
    }

    /**
     * 获取锁
     * @param lockPath
     */
    public static void getLock(String lockPath){
        // 创建有序临时目录
        final String ephemeralSequential1 = zkClient.createEphemeralSequential(LOCK_PARENT_PATH + "/" + lockPath, new Byte[0]);
        List<String> subNodes = zkClient.getChildren(LOCK_PARENT_PATH);
        if(ephemeralSequential1.equals(subNodes.get(0))){
            // 当前为第一位有序目录-获取到锁
            return;
        }
        final Object obj = Thread.currentThread();
        zkClient.subscribeChildChanges(LOCK_PARENT_PATH, new IZkChildListener() {
            public void handleChildChange(String s, List<String> list) throws Exception {
                if(list != null
                        && list.size() !=0
                        && (LOCK_PARENT_PATH + "/" + list.get(0)).equals(ephemeralSequential1)){
                    synchronized (obj){
                        obj.notify();
                    }
                }
            }
        });
        synchronized (obj){
            try {
                obj.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }


    /**
     * 还锁
     * @param lockPath
     */
    public static void alsoLock(String lockPath){
        zkClient.delete(LOCK_PARENT_PATH + "/" +lockPath);
    }

    /**
     * 测试用例
     * @param args
     * @throws InterruptedException
     */
    public static void main(String [] args) throws InterruptedException {
        final ZkClient zkClient = new ZkClient(servers, 10000);
        boolean panss = zkClient.exists("/sys/app");
        if(panss){
            zkClient.deleteRecursive("/sys/app");
        }
        zkClient.createPersistent("/sys/app");
        final String ephemeralSequential1 = zkClient.createEphemeralSequential("/sys/app/lock", new Byte[0]);
        final String ephemeralSequential2 = zkClient.createEphemeralSequential("/sys/app/lock", new Byte[0]);
        List<String> subNodes = zkClient.getChildren("/sys/app");
        System.out.println("当前子节点情况：" + JSON.toJSONString(subNodes));
        final Object obj = Thread.currentThread();
        // 监听子节点变化
        final Long currentThreadId = Thread.currentThread().getId();
        System.out.println("当前主线程id:"+Thread.currentThread().getId());
        zkClient.subscribeChildChanges("/sys/app", new IZkChildListener() {
            public void handleChildChange(String s, List<String> list) throws Exception {
                System.out.println("节点发生变化：" + s + "当前线程id："+Thread.currentThread().getId() + " list:" + JSON.toJSONString(list));
                if(list != null && list.size() !=0 && ("/sys/app/" + list.get(0)).equals(ephemeralSequential2)){
                    ThreadMXBean tmx = ManagementFactory.getThreadMXBean();
                    ThreadInfo info = tmx.getThreadInfo(currentThreadId);
                    System.out.println("主线程状态："+info.getThreadState());
                    synchronized (obj){
                        obj.notify();
                    }
                }
            }
        });
        new Thread(new Runnable() {
            public void run() {
                try {
                    System.out.println("等待三秒后释放锁");
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                zkClient.delete(ephemeralSequential1);
            }
        }).start();
        System.out.println("主线程等待获取到锁");
        synchronized (obj){
            obj.wait();
        }
        System.out.println("主线程被唤醒，获取到锁了");
    }
}
